package com.atguigu.gmall.realtime.dim.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.TableProcessDim;
import com.atguigu.gmall.realtime.common.util.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.*;

/**
 * ClassName: TableProcessFunction
 * Description:
 * 处理主流、广播流--
 * @Create 2024/4/17 14:18
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>{

    Map<String, TableProcessDim> configMap = new HashMap<>();

    MapStateDescriptor<String, TableProcessDim> mapStateDescriptor;

    public TableProcessFunction(MapStateDescriptor<String, TableProcessDim> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        Connection conn = JdbcUtil.getMysqlConnection();
        String sql = "select * from gmall1030_config.table_process_dim";
        List<TableProcessDim> tableProcessDims = JdbcUtil.queryList(conn, sql, TableProcessDim.class, true);
        for (TableProcessDim tableProcessDim : tableProcessDims) {
            configMap.put(tableProcessDim.getSourceTable(), tableProcessDim);
        }
        JdbcUtil.closeConnection(conn);

    }

    @Override
    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.ReadOnlyContext readOnlyContext, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
        ReadOnlyBroadcastState<String, TableProcessDim> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);
        //从当前处理的业务数据中获取表名
        String table = jsonObject.getString("table");

        //根据表名从广播状态获取对应的配置对象
        TableProcessDim tableProcessDim = broadcastState.get(table);

        //如果对象不为空，说明是维度数据，将其中data部分传递到下游
        if ((tableProcessDim = broadcastState.get(table)) != null || (tableProcessDim = configMap.get(table)) != null) {
            JSONObject dataJsonObj = jsonObject.getJSONObject("data");

            String sinkColumns = tableProcessDim.getSinkColumns();
            deleteNotNeedColums(dataJsonObj,sinkColumns);


            //在向下游传递数据前，补充对维度表数据的的操作类型
            String type = jsonObject.getString("type");
            dataJsonObj.put("type", type);

            //将当前处理的维度数据以及这条数据对应的配置文件封装为Tuple2对象，传递到下游
            collector.collect(Tuple2.of(dataJsonObj,tableProcessDim));
        }

    }

    //2024-04-17测试，业务数据流未生成数据，而广播数据流默认会读取历史数据，因此广播流数据先到
    //2024-04-17解决广播流和数据流先后问题：open方法
    @Override
    public void processBroadcastElement(TableProcessDim tableProcessDim, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.Context context, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
        //获取广播状态
        BroadcastState<String, TableProcessDim> broadcastState = context.getBroadcastState(mapStateDescriptor);

        //获取对配置表的操作类型，根据操作类型
        String op = tableProcessDim.getOp();

        String key = tableProcessDim.getSourceTable();

        if ("d".equals(op)) {
            //说明从配置表中删除一条数据，将这条数据对应的配置信息从广播状态中删除
            broadcastState.remove(key);
        } else {
            //说明对配置表进行了C、R、U，将这条数据对应的配置信息添加到广播状态中
            broadcastState.put(key, tableProcessDim);
        }
    }

    //过滤不需要传递的字段
    //不需要返回值，引用地址，entries操作的也是原对象
    private static void deleteNotNeedColums(JSONObject dataJsonObj,String sinkColumns) {

        List<String> columnList = Arrays.asList(sinkColumns.split(","));

        Set<Map.Entry<String, Object>> entries = dataJsonObj.entrySet();

        entries.removeIf(entry -> !columnList.contains(entry.getKey()));
    }

}



